"""
Case Type   : ustore_DQL_SET_OPERATIONS
Case Name   : with在主查询中通过名称引用的子查询
Create At   : 2022-05-12
Owner       : nanyang12
Description :
    1.创建表
    2.使用copy命令插入数据
    3.在主查询中通过名称引用的子查询
    4.环境清理
Expect      :
    1.创建成功
    2.数据插入成功
    3.成功
    4.清理成功
History     :
"""

import os
import subprocess
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Ustore01(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----- {os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_root = Node('PrimaryRoot')
        self.com = Common()
        self.constant = Constant()
        self.table_name1 = 't_ustore_winagg_0002_01'
        self.table_name2 = 't_ustore_winagg_0002_02'
        self.table_name3 = 't_ustore_winagg_0002_03'
        self.table_name4 = 't_ustore_winagg_0002_04'
        self.table_name5 = 't_ustore_winagg_0002_05'
        self.copy_file1 = 'warehouse_33.txt'
        self.copy_file2 = 'web_site_33.txt'
        self.copy_file3 = 'store_returns_33.txt'
        self.copy_file4 = 'customer_33.txt'
        self.copy_file5 = 'date_dim_33.txt'
        self.cur_path_1 = os.path.join(macro.SCRIPTS_PATH, self.copy_file1)
        self.cur_path_2 = os.path.join(macro.SCRIPTS_PATH, self.copy_file2)
        self.cur_path_3 = os.path.join(macro.SCRIPTS_PATH, self.copy_file3)
        self.cur_path_4 = os.path.join(macro.SCRIPTS_PATH, self.copy_file4)
        self.cur_path_5 = os.path.join(macro.SCRIPTS_PATH, self.copy_file5)
        self.parent_path = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.target_path1 = os.path.join(self.parent_path, self.copy_file1)
        self.target_path2 = os.path.join(self.parent_path, self.copy_file2)
        self.target_path3 = os.path.join(self.parent_path, self.copy_file3)
        self.target_path4 = os.path.join(self.parent_path, self.copy_file4)
        self.target_path5 = os.path.join(self.parent_path, self.copy_file5)

    def test_join(self):
        step_txt = '-----step1:创建表;expect: 创建成功-----'
        self.log.info(step_txt)
        cmd_sql = f"""drop table if exists {self.table_name1}  CASCADE ;
                    drop table if exists {self.table_name2}  CASCADE ;
                    drop table if exists {self.table_name3}  CASCADE ;
                    drop table if exists {self.table_name4}  CASCADE ;
                    drop table if exists {self.table_name5}  CASCADE ;
            create table {self.table_name1} 
            (
                d_date_sk                 integer               not null,
                d_date_id                 char(16)              not null,
                d_date                    date                          ,
                d_month_seq               integer                       ,
                d_week_seq                integer                       ,
                d_quarter_seq             integer                       ,
                d_year                    integer                       ,
                d_dow                     integer                       ,
                d_moy                     integer                       ,
                d_dom                     integer                       ,
                d_qoy                     integer                       ,
                d_fy_year                 integer                       ,
                d_fy_quarter_seq          integer                       ,
                d_fy_week_seq             integer                       ,
                d_day_name                char(9)                       ,
                d_quarter_name            char(6)                       ,
                d_holiday                 char(1)                       ,
                d_weekend                 char(1)                       ,
                d_following_holiday       char(1)                       ,
                d_first_dom               integer                       ,
                d_last_dom                integer                       ,
                d_same_day_ly             integer                       ,
                d_same_day_lq             integer                       ,
                d_current_day             char(1)                       ,
                d_current_week            char(1)                       ,
                d_current_month           char(1)                       ,
                d_current_quarter         char(1)                       ,
                d_current_year            char(1)                       
             )
            WITH (STORAGE_TYPE=USTORE) partition by Range(d_year) (
             partition p1 values less than(1950),
             partition p2 values less than(2000),
             partition p3 values less than(2050),
             partition p4 values less than(2100),
             partition p5 values less than(3000),
             partition p6 values less than(maxvalue)
            );
            create table {self.table_name2}
            (
                s_store_sk                integer               not null,
                s_store_id                char(16)              not null,
                s_rec_start_date          date                          ,
                s_rec_end_date            date                          ,
                s_closed_date_sk          integer                       ,
                s_store_name              varchar(50)                   ,
                s_number_employees        integer                       ,
                s_floor_space             integer                       ,
                s_hours                   char(20)                      ,
                s_manager                 varchar(40)                   ,
                s_market_id               integer                       ,
                s_geography_class         varchar(100)                  ,
                s_market_desc             varchar(100)                  ,
                s_market_manager          varchar(40)                   ,
                s_division_id             integer                       ,
                s_division_name           varchar(50)                   ,
                s_company_id              integer                       ,
                s_company_name            varchar(50)                   ,
                s_street_number           varchar(10)                   ,
                s_street_name             varchar(60)                   ,
                s_street_type             char(15)                      ,
                s_suite_number            char(10)                      ,
                s_city                    varchar(60)                   ,
                s_county                  varchar(30)                   ,
                s_state                   char(2)                       ,
                s_zip                     char(10)                      ,
                s_country                 varchar(20)                   ,
                s_gmt_offset              decimal(5,2)                  ,
                s_tax_precentage          decimal(5,2)                  
             )  WITH (STORAGE_TYPE=USTORE);
        create table {self.table_name3}
        (
            sr_returned_date_sk       integer                       ,
            sr_return_time_sk         integer                       ,
            sr_item_sk                integer               not null,
            sr_customer_sk            integer                       ,
            sr_cdemo_sk               integer                       ,
            sr_hdemo_sk               integer                       ,
            sr_addr_sk                integer                       ,
            sr_store_sk               integer                       ,
            sr_reason_sk              integer                       ,
            sr_ticket_number          bigint               not null,
            sr_return_quantity        integer                       ,
            sr_return_amt             decimal(7,2)                  ,
            sr_return_tax             decimal(7,2)                  ,
            sr_return_amt_inc_tax     decimal(7,2)                  ,
            sr_fee                    decimal(7,2)                  ,
            sr_return_ship_cost       decimal(7,2)                  ,
            sr_refunded_cash          decimal(7,2)                  ,
            sr_reversed_charge        decimal(7,2)                  ,
            sr_store_credit           decimal(7,2)                  ,
            sr_net_loss               decimal(7,2)                  
         )
        WITH (STORAGE_TYPE=USTORE) partition by range(sr_returned_date_sk)
        (
         partition p1 values less than (2451179) ,
         partition p2 values less than (2451544) ,
         partition p3 values less than (2451910) ,
         partition p4 values less than (2452275) ,
         partition p5 values less than (2452640) ,
         partition p6 values less than (2453005) ,
         partition p7 values less than (maxvalue)
        );
        create table {self.table_name4} 
        (
            c_customer_sk             integer               not null,
            c_customer_id             char(16)              not null,
            c_current_cdemo_sk        integer                       ,
            c_current_hdemo_sk        integer                       ,
            c_current_addr_sk         integer                       ,
            c_first_shipto_date_sk    integer                       ,
            c_first_sales_date_sk     integer                       ,
            c_salutation              char(10)                      ,
            c_first_name              char(20)                      ,
            c_last_name               char(30)                      ,
            c_preferred_cust_flag     char(1)                       ,
            c_birth_day               integer                       ,
            c_birth_month             integer                       ,
            c_birth_year              integer                       ,
            c_birth_country           varchar(20)                   ,
            c_login                   char(13)                      ,
            c_email_address           char(50)                      ,
            c_last_review_date        char(10)
        )  WITH (STORAGE_TYPE=USTORE);
        create table {self.table_name5} 
        (
            d_date_sk                 integer               not null,
            d_date_id                 char(16)              not null,
            d_date                    date                          ,
            d_month_seq               integer                       ,
            d_week_seq                integer                       ,
            d_quarter_seq             integer                       ,
            d_year                    integer                       ,
            d_dow                     integer                       ,
            d_moy                     integer                       ,
            d_dom                     integer                       ,
            d_qoy                     integer                       ,
            d_fy_year                 integer                       ,
            d_fy_quarter_seq          integer                       ,
            d_fy_week_seq             integer                       ,
            d_day_name                char(9)                       ,
            d_quarter_name            char(6)                       ,
            d_holiday                 char(1)                       ,
            d_weekend                 char(1)                       ,
            d_following_holiday       char(1)                       ,
            d_first_dom               integer                       ,
            d_last_dom                integer                       ,
            d_same_day_ly             integer                       ,
            d_same_day_lq             integer                       ,
            d_current_day             char(1)                       ,
            d_current_week            char(1)                       ,
            d_current_month           char(1)                       ,
            d_current_quarter         char(1)                       ,
            d_current_year            char(1)                       
         )
        WITH (STORAGE_TYPE=USTORE) partition by Range(d_year) (
         partition p1 values less than(1950),
         partition p2 values less than(2000),
         partition p3 values less than(2050),
         partition p4 values less than(2100),
         partition p5 values less than(3000),
         partition p6 values less than(maxvalue)
        );"""
        self.log.info(cmd_sql)
        create_result = self.pri_sh.execut_db_sql(cmd_sql)
        self.log.info(create_result)
        self.assertTrue(
            create_result.count(self.constant.TABLE_CREATE_SUCCESS) == 5,
            '执行失败:' + step_txt)

        step_txt = '-----step2:使用copy命令插入数据，expect: 数据插入成功-----'
        self.log.info(step_txt)
        text = '-----step2.1: 获取当前主机的IP地址;expect: 成功-----'
        self.log.info(text)
        cmd = "ifconfig -a|grep inet6 -a2|grep broadcast|awk '{print $2}'"
        self.log.info(cmd)
        self.cur_ip = subprocess.getoutput(cmd).strip()
        self.log.info(self.cur_ip)

        text = '-----step2.2: 复制文件到数据库主机对应目录 expect: 成功-----'
        self.log.info(text)
        if self.cur_ip == self.pri_sh.node.ssh_host:
            self.log.info('当前主机是数据库主机，无需远程复制')
            cmd = f"cp {self.cur_path_1} {self.parent_path};" \
                f"cp {self.cur_path_2} {self.parent_path};" \
                f"cp {self.cur_path_3} {self.parent_path};" \
                f"cp {self.cur_path_4} {self.parent_path};" \
                f"cp {self.cur_path_5} {self.parent_path};"
            self.log.info(cmd)
            res = self.pri_sh.node.sh(cmd).result()
            self.log.info(res)
        else:
            self.com.scp_file(self.pri_root, self.copy_file1,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file2,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file3,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file4,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file5,
                              self.parent_path)
            chmod_cmd = f"chmod 777 -R {self.parent_path};" \
                        f"ls -al {self.parent_path}"
            self.log.info(chmod_cmd)
            chmod_result = self.pri_root.sh(chmod_cmd).result()
            self.log.info(chmod_result)

        text = '-----step2.3: 查看数据库主机目标路径下文件 expect: 成功-----'
        self.log.info(text)
        cmd = f"ls -l {self.parent_path}| grep -E '{self.copy_file1}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file2}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file3}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file4}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file5}';"
        self.log.info(cmd)
        res = self.pri_root.sh(cmd).result()
        self.log.info(res)
        self.assertNotIn('No such file or directory', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file1}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file2}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file3}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file4}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file5}', res, '文件拷贝失败')

        text = '-----step2.4:使用copy命令插入数据，expect: 数据插入成功-----'
        self.log.info(text)
        self.log.info('---------------使用copy命令插入数据------------------')
        copy_sql = f"copy {self.table_name1} from '{self.target_path1}' " \
            f"delimiters ',';" \
            f"copy {self.table_name2} from '{self.target_path2}' " \
            f"delimiters ',';" \
            f"copy {self.table_name3} from '{self.target_path3}' " \
            f"delimiters ',';" \
            f"copy {self.table_name4} from '{self.target_path4}' " \
            f"delimiters ',';" \
            f"copy {self.table_name5} from '{self.target_path5}' " \
            f"delimiters ',';"
        self.log.info(copy_sql)
        copy_result = self.pri_sh.execut_db_sql(copy_sql)
        self.log.info(copy_result)
        self.assertIn('COPY 10000', copy_result, '执行失败:' + step_txt)
        self.assertIn('COPY 2875', copy_result, '执行失败:' + step_txt)

        step_txt = '-----step3: union在OALP场景测试 expect: 成功-----'
        self.log.info(step_txt)
        cmd_sql = f"""
with customer_total_return as
 (select sr_customer_sk as ctr_customer_sk,
         sr_store_sk as ctr_store_sk,
         sum((sr_return_amt)) over(partition by avg((select sr_return_amt
              from {self.table_name3}
             where sr_returned_date_sk in
                   (select d_date_sk
                      from {self.table_name5}
                     where d_year = 2000
                     except
                      select s_closed_date_sk
                              from {self.table_name2}
                             group by s_closed_date_sk
                    )
               and sr_returned_date_sk =
                   2451179
             group by sr_return_amt
             order by 1 limit 1)) order by sr_refunded_cash) as ctr_total_return
    from {self.table_name3}, {self.table_name1}
   where sr_returned_date_sk = d_date_sk
     and d_year = 2000
     and d_moy < 2
   group by sr_customer_sk, sr_store_sk, sr_return_amt, sr_refunded_cash)
select c_customer_id, ctr_total_return
  from customer_total_return ctr1, {self.table_name2}, {self.table_name4}
 where exists (select avg(ctr_total_return) over
 (partition by ctr_store_sk order by ctr_customer_sk)
          from customer_total_return ctr2
         where ctr1.ctr_store_sk = ctr2.ctr_store_sk)
   and s_store_sk = ctr1.ctr_store_sk
   and s_state = 'TN'
   and ctr1.ctr_customer_sk = c_customer_sk
   and substr(c_customer_sk, -2) = 11
 order by c_customer_id, 2 limit 20;"""
        self.log.info(cmd_sql)
        query_result = self.pri_sh.execut_db_sql(cmd_sql)
        self.log.info(query_result)
        self.assertIn('0 rows', query_result, '执行失败:' + step_txt)

    def tearDown(self):
        step_txt = '-----step4: 环境清理 expect:清理成功-----'
        self.log.info(step_txt)
        self.log.info('删除表')
        cmd_sql = self.pri_sh.execut_db_sql(f"""
                    drop table {self.table_name1} cascade;
                    drop table {self.table_name2} cascade;
                    drop table {self.table_name3} cascade;
                    drop table {self.table_name4} cascade;
                    drop table {self.table_name5} cascade;""")
        self.log.info(cmd_sql)
        text = '-----删除文件-----'
        self.log.info(text)
        rm_cmd = f"rm -rf {self.target_path1};" \
            f"rm -rf {self.target_path2};" \
            f"rm -rf {self.target_path3};" \
            f"rm -rf {self.target_path4};" \
            f"rm -rf {self.target_path5};"
        self.log.info(rm_cmd)
        self.com.get_sh_result(self.pri_root, rm_cmd)
        for s in [self.target_path1, self.target_path2]:
            cmd = f'if [ -f {s} ]; then echo "does exists"; ' \
                f'else echo "not exists"; fi'
            file_judgments = self.com.get_sh_result(self.pri_root, cmd)
            self.assertIn('not exists', file_judgments, "清理失败" + text)
        self.assertTrue(
            cmd_sql.count(self.constant.DROP_TABLE_SUCCESS) == 5,
            '执行失败:' + step_txt)
        self.log.info(f'----- {os.path.basename(__file__)} end-----')
